package com.atguigu.gmall.realtime.dws.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.base.BaseApp;
import com.atguigu.gmall.realtime.common.bean.TradePaymentBean;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.function.DorisMapFunction;
import com.atguigu.gmall.realtime.common.util.DateFormatUtil;
import com.atguigu.gmall.realtime.common.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

public class DwsTradePaymentSucWindow extends BaseApp {
      // 主方法，应用程序入口
      public static void main(String[] args) {
// 启动应用，传入端口号、并行度、表名和 Kafka 主题
            new DwsTradePaymentSucWindow().start(
                        10027, // 端口号
                        4, // 并行度
                        "dws_trade_payment_suc_window", // 表名
                        Constant.TOPIC_DWD_TRADE_ORDER_PAYMENT_SUCCESS // Kafka 主题
            );
      }
      @Override
      public void handle(StreamExecutionEnvironment env, DataStreamSource<String>
                  stream) {
            stream
                        .map(JSON::parseObject) // 将每条 JSON 字符串解析为 JSONObject
                        .keyBy(obj -> obj.getString("user_id")) // 按用户 ID 分组
                        .process(new KeyedProcessFunction<String, JSONObject,
                                    TradePaymentBean>() {
                              private ValueState<String> lastPayDateState; // 状态保存上一次支付的日期
                              @Override
                              public void open(Configuration parameters) throws Exception {
// 初始化状态
                                    lastPayDateState = getRuntimeContext().getState(new
                                                ValueStateDescriptor<>("lastPayDate", String.class));
                              }
                              @Override
                              public void processElement(JSONObject obj, Context ctx,
                                                         Collector<TradePaymentBean> out) throws Exception {
                                    String lastPayDate = lastPayDateState.value(); // 获取状态中的上一次支付日期
                                    long ts = obj.getLong("ts") * 1000; // 获取事件时间戳并转换为毫秒
                                    String today = DateFormatUtil.tsToDate(ts); // 将时间戳转换为日期字符串
                                    long payUuCount = 0L; // 记录支付成功的唯一用户数量
                                    long payNewCount = 0L; // 记录支付成功的新用户数量
                                    if (!today.equals(lastPayDate)) { // 如果今天与状态中的日期不同，则说明是当天第一次支付成功
                                          lastPayDateState.update(today); // 更新状态中的支付日期
                                          payUuCount = 1L; // 记录唯一用户支付成功
                                          if (lastPayDate == null) {
// 如果状态中的最后支付日期为 null，说明该用户之前没有支付过，为新用户支付
                                                            payNewCount = 1L;
                                          }
                                    }
                                    if (payUuCount == 1) {
// 只有在 payUuCount 为 1 时，才输出数据
                                          out.collect(new TradePaymentBean("", "", "", payUuCount,
                                                      payNewCount, ts));
                                    }
                              }
                        })
                        .assignTimestampsAndWatermarks(
                                    WatermarkStrategy
                                                .
                                                <TradePaymentBean>forBoundedOutOfOrderness(Duration.ofSeconds(5L)) // 设置水印策略，允许最大 5 秒的乱序
                                                .withTimestampAssigner((bean, ts) -> bean.getTs()) // 使用`ts` 字段作为事件时间戳
                                                .withIdleness(Duration.ofSeconds(120L)) // 设置空闲时间，如果120 秒内没有数据则认为窗口处于空闲状态
                        )
                        .windowAll(TumblingEventTimeWindows.of(Time.seconds(5L))) // 使用 5 秒的滚动时间窗口
                        .reduce(
                                    new ReduceFunction<TradePaymentBean>() {
                                          @Override
                                          public TradePaymentBean reduce(TradePaymentBean value1,
                                                                         TradePaymentBean value2) {
// 合并窗口内的 TradePaymentBean 对象的计数值
                                                value1.setPaymentSucNewUserCount(value1.getPaymentSucNewUserCount() +
                                                            value2.getPaymentSucNewUserCount());
                                                value1.setPaymentSucUniqueUserCount(value1.getPaymentSucUniqueUserCount() +
                                                            value2.getPaymentSucUniqueUserCount());
                                                return value1;
                                          }
                                    },
                                    new ProcessAllWindowFunction<TradePaymentBean, TradePaymentBean,
                                                TimeWindow>() {
                                          @Override
                                          public void process(Context ctx, Iterable<TradePaymentBean>
                                                      elements, Collector<TradePaymentBean> out) throws Exception {
                                                TradePaymentBean bean = elements.iterator().next(); //获取窗口内的聚合结果
// 将处理结果封装到 TradePaymentBean 对象中，并输出
                                                bean.setStt(DateFormatUtil.tsToDateTime(ctx.window().getStart())); // 窗口起始时间
                                                bean.setEdt(DateFormatUtil.tsToDateTime(ctx.window().getEnd())); // 窗口结束时间
                                                bean.setCurDate(DateFormatUtil.tsToDateForPartition(ctx.window().getStart()));
// 当前日期
                                                out.collect(bean);
                                          }
                                    }
                        )
                        .map(new DorisMapFunction<>()) // 将数据转换为 Doris 所需的格式
                        .sinkTo(FlinkSinkUtil.getDorisSink(
                                    Constant.DORIS_DATABASE + ".dws_trade_payment_suc_window", //Doris 数据库表名
                                    "dws_trade_payment_suc_window" // Doris 表名
                        ));
      }
}
